package com.studerb.service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.time.StopWatch;
import org.apache.log4j.Logger;
import org.bushe.swing.event.annotation.AnnotationProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import com.studerb.dao.interfaces.IHeaderDao;
import com.studerb.dao.interfaces.IHeaderThreadDao;
import com.studerb.events.HeaderEvent;
import com.studerb.events.NewsGroupEvent;
import com.studerb.exceptions.ServiceException;
import com.studerb.model.Header;
import com.studerb.model.HeaderThread;
import com.studerb.model.NewsGroup;
import com.studerb.model.Post;
import com.studerb.service.interfaces.IHeaderService;
import com.studerb.service.interfaces.IService;
import com.studerb.service.interfaces.ISettingsService;
import com.studerb.service.interfaces.ITaskService;
import com.studerb.tasks.TaskType;
import com.studerb.tasks.headers.ArticleCommandTask;
import com.studerb.tasks.headers.BodyCommandTask;
import com.studerb.tasks.headers.XoverAllCommandTask;
import com.studerb.tasks.headers.XoverNewCommandTask;
import com.studerb.tasks.interfaces.ITaskFactory;
import com.studerb.tasks.interfaces.ITaskListener;
import com.studerb.util.Constants;

public class HeaderService implements IHeaderService, IService {
	private final static Logger logger = Logger.getLogger(HeaderService.class);
	private boolean started = false;
	private int DB_BATCH_SIZE;
	@Autowired protected ITaskFactory taskFactory;
	@Autowired protected ITaskService taskService;
	@Autowired protected EventService eventService;
	@Autowired protected IHeaderDao headerDao;
	@Autowired protected IHeaderThreadDao headerThreadDao;
	@Autowired protected ISettingsService settingsService;
	@Autowired protected PlatformTransactionManager transactionManager;

	private TransactionTemplate transactionTemplate;
	private final Pattern subjectPattern = Pattern.compile("^(.*)([\\(\\[](\\d+)\\/(\\d+)[\\)\\]])(.*)");

	private void _updateHeader(Header header) {
		headerDao.update(header);
		eventService.publish(new HeaderEvent(HeaderEvent.EventType.UPDATED, header));
	}

	private void createHeaderThreads(List<Header> headers, NewsGroup newsGroup) {
		SubjectComparator subjectComparator = new SubjectComparator();
		List<HeaderThread> headerThreads = headerThreadDao.getIncomplete(newsGroup, null, null);
		for (Header header : headers) {
			ParsedSubject parsedSubject = parseSubject(header);
			if (parsedSubject == null) {
				continue;
			}
			header.setCurrentPart(parsedSubject.getCurrentPart());
			HeaderThread headerThread = new HeaderThread(header, parsedSubject);
			int index = Collections.binarySearch(headerThreads, headerThread, subjectComparator);

			if (index >= 0) {
				headerThread = headerThreads.get(index);
			}
			else {
				headerThreads.add(-1 * index - 1, headerThread);
			}
			headerThread.addHeader(header);
			header.setHeaderThread(headerThread);
			if (headerThread.getArticleNumber() == null) {
				logger.warn("SAVING null articleNumber: " + headerThread.toString());
			}
			headerThreadDao.saveOrUpdate(headerThread);
			if (headerThread.isComplete()) {
				headerThreads.remove(headerThread);
			}
			header.setHeaderThread(headerThread);
		}
	}

	@Transactional @Override public int deleteAllPosts(NewsGroup newsGroup) {
		int deletedHeaders = headerDao.deleteByNewsGroup(newsGroup, null, null);
		int deletedThreads = headerThreadDao.deleteByNewsGroup(newsGroup, null, null);
		eventService.publish(new NewsGroupEvent(NewsGroupEvent.EventType.HEADERS, newsGroup));
		return deletedHeaders + deletedThreads;
	}

	@Override public void downloadAllHeaders(NewsGroup group, ITaskListener... taskListeners) throws ServiceException {
		XoverAllCommandTask xoverTask = (XoverAllCommandTask) taskFactory.getTask(TaskType.XOVER_ALL);
		xoverTask.setNewsGroup(group);
		for (ITaskListener listener : taskListeners) {
			xoverTask.addTaskListener(listener);
		}
		boolean success = taskService.addTask(xoverTask);
		if (!success) {
			throw new ServiceException("Error adding Task to task service");
		}
	}

	@Override public void downloadArticle(Post post, ITaskListener... taskListeners) throws ServiceException {
		ArticleCommandTask task = (ArticleCommandTask) taskFactory.getTask(TaskType.ARTICLE);
		task.setPost(post);
		for (ITaskListener taskListener : taskListeners) {
			task.addTaskListener(taskListener);
		}
		if (!taskService.addTask(task)) {
			throw new ServiceException("Error submitting ARTICLE task to TaskService");
		}
	}

	@Override public void downloadBody(Post post, ITaskListener... taskListeners) throws ServiceException {
		BodyCommandTask task = (BodyCommandTask) taskFactory.getTask(TaskType.BODY);
		task.setPost(post);
		for (ITaskListener taskListener : taskListeners) {
			task.addTaskListener(taskListener);
		}
		if (!taskService.addTask(task)) {
			throw new ServiceException("Error submitting BODY task to TaskService");
		}
	}

	@Override public void downloadHeaders(NewsGroup group, int first, int last, ITaskListener... taskListeners) throws ServiceException {
		throw new UnsupportedOperationException("Method not implemented yet");
	}

	@Override public void downloadNewHeaders(NewsGroup group, ITaskListener... taskListeners) throws ServiceException {
		XoverNewCommandTask xoverTask = (XoverNewCommandTask) taskFactory.getTask(TaskType.XOVER_NEW);
		xoverTask.setNewsGroup(group);
		for (ITaskListener listener : taskListeners) {
			xoverTask.addTaskListener(listener);
		}
		boolean success = taskService.addTask(xoverTask);
		if (!success) {
			throw new ServiceException("Error adding Task to task service");
		}
	}

	@Transactional(readOnly = true) @Override public List<Header> getHeadersByThread(HeaderThread headerThread) {
		return headerThreadDao.getHeadersByThread(headerThread);
	}

	@Transactional(readOnly = true) @Override public List<Post> getPosts(NewsGroup newsGroup, Long min, Long max) {
		StopWatch stopWatch = new StopWatch();
		logger.debug("getting posts for group: " + newsGroup + " between " + min + " / " + max);

		stopWatch.start();
		List<HeaderThread> headerThreads = headerThreadDao.getHeaderThreads(newsGroup, min, max);
		stopWatch.stop();
		logger.debug("TIME: Getting headerThreads: " + stopWatch.toString());

		stopWatch.reset();
		stopWatch.start();
		List<Header> headers = headerDao.getAllSinglePart(newsGroup, min, max);
		stopWatch.stop();
		logger.debug("TIME: Getting singleHeaders: " + stopWatch.toString());

		stopWatch.reset();
		stopWatch.start();
		List<Post> posts = new ArrayList<Post>(headerThreads.size() + headers.size());
		posts.addAll(headerThreads);
		posts.addAll(headers);
		Collections.sort(posts);
		stopWatch.stop();
		logger.debug("TIME: Adding posts and sorting: " + stopWatch.toString());

		return posts;
	}

	@Override public Callable<Void> getXoverCommandProcessor(BlockingQueue<String> queue, NewsGroup newsGroup) {
		return new XoverCommandProcessor(queue, newsGroup);
	}

	@Override public boolean isStarted() {
		return started;
	}

	private ParsedSubject parseSubject(Header header) {
		String subject = header.getSubject();
		Matcher matcher = subjectPattern.matcher(subject);
		if (matcher.find()) {
			String bareSubject = subject.replace(matcher.group(2), "") + matcher.group(5);
			Integer current = Integer.parseInt(matcher.group(3));
			Integer max = Integer.parseInt(matcher.group(4));
			if (current <= max) {
				return new ParsedSubject(bareSubject, current, max);
			}
		}
		return null;
	}

	@Transactional(propagation = Propagation.REQUIRED) public void saveHeaders(final List<Header> headers, final NewsGroup newsGroup) {
		final Long min = headers.get(0).getArticleNumber();
		final Long max = headers.get(headers.size() - 1).getArticleNumber();
		// headerDao.deleteByNewsGroup(newsGroup, min, max);
		// headerThreadDao.deleteByNewsGroup(newsGroup, min, max);
		createHeaderThreads(headers, newsGroup);
		headerDao.saveHeaders(headers, newsGroup);
		eventService.publish(new NewsGroupEvent(NewsGroupEvent.EventType.HEADERS, newsGroup));
	}

	@Transactional @Override public void setArticleDownloaded(Post post) {
		post.setArticleDownloaded(true);
		if (post instanceof Header) {
			_updateHeader((Header) post);
		}
		else if (post instanceof HeaderThread) {
			headerThreadDao.update((HeaderThread) post);
		}
	}

	@Override public void start() {
		if (started) {
			throw new RuntimeException("HeaderService already Started");
		}
		transactionTemplate = new TransactionTemplate(transactionManager);
		logger.info("[**HeaderService Starting**]");
		DB_BATCH_SIZE = Integer.parseInt(settingsService.getSetting(Constants.DB_BATCH_SIZE));
		AnnotationProcessor.process(this);
		started = true;

	}

	@Override public void stop() {
		if (!started) {
			throw new RuntimeException("HeaderService already Stopped");
		}
		logger.info("[**HeaderService Stopping**]");
		started = false;
	}

	class XoverCommandProcessor implements Callable {
		private final BlockingQueue<String> queue;
		private final NewsGroup newsGroup;

		public XoverCommandProcessor(BlockingQueue<String> queue, NewsGroup newsGroup) {
			this.queue = queue;
			this.newsGroup = newsGroup;
		}

		public Object call() throws Exception {
			logger.debug("XoverCommandProcessor starting...");
			final List<String> badHeaders = new ArrayList<String>();
			final List<Header> tempHeaders = new ArrayList<Header>();
			int totalLines = 0;
			int goodLines = 0;
			String nextLine = null;
			Long lastArticleNumber = new Long(-1);
			try {
				do {
					nextLine = queue.take();
					totalLines++;
					Header tempHeader = Header.createHeader(nextLine, newsGroup);
					if (tempHeader != null) {
						goodLines++;
						tempHeaders.add(tempHeader);
						if (tempHeader.getArticleNumber() <= lastArticleNumber) {
							throw new RuntimeException("Error: last downloaded article number greater than previous...");
						}
					}
					else if (!nextLine.equals(Constants.END_OF_QUEUE)) {
						badHeaders.add(nextLine);
					}

					if (!tempHeaders.isEmpty()) {
						if (tempHeaders.size() % DB_BATCH_SIZE == 0 || nextLine.equals(Constants.END_OF_QUEUE)) {
							transactionTemplate.execute(new TransactionCallbackWithoutResult() {
								@Override protected void doInTransactionWithoutResult(TransactionStatus arg0) {
									saveHeaders(tempHeaders, newsGroup);
								}
							});

							logger.debug("saved/updated " + tempHeaders.size() + " headers");
							tempHeaders.clear();
						}
					}
				}
				while (!nextLine.equals(Constants.END_OF_QUEUE));
			}
			catch (InterruptedException interrupted) {
				logger.debug("Consumer interrupted - returning");
				return null;
			}
			finally {
				queue.clear();
			}
			logger.debug("***Bad Headers: ");
			for (String badHeader : badHeaders) {
				logger.warn("\t\t" + badHeader);
			}
			logger.info(String.format("XOverConsumer: processed: %d / %d for group: %s", goodLines, totalLines, newsGroup.getGroupName()));
			return null;
		}
	}
}

class SubjectComparator implements Comparator<HeaderThread> {
	public int compare(HeaderThread o1, HeaderThread o2) {
		return o1.getSubject().compareToIgnoreCase(o2.getSubject());
	}
}
